package com.acme.multithreadinglab.JCIPexample;

import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.*;

public class ProducerConsumer {

	static class FileCrawler implements Runnable {
		private final List<BlockingDeque<File>> fileQueue;
		private final FileFilter fileFilter;
		private final File root;
		private int queueIndex;
		private CountDownLatch stopLatch;

		public FileCrawler(List<BlockingDeque<File>> fileQueue,
				final FileFilter fileFilter, File root, CountDownLatch stopLatch) {
			this.fileQueue = fileQueue;
			this.root = root;
			this.fileFilter = new FileFilter() {
				public boolean accept(File f) {
					return f.isDirectory() || fileFilter.accept(f);
				}
			};
			this.stopLatch = stopLatch;
		}

		private boolean alreadyIndexed(File f) {
			return false;
		}

		public void run() {
			try {
				crawl(root);
			} catch (InterruptedException e) {
				Thread.currentThread().interrupt();
			}
			stopLatch.countDown();
			System.out.println("Crawler Thread " + Thread.currentThread().getId() + " interrupted");
		}

		private void crawl(File root) throws InterruptedException {
			File[] entries = root.listFiles(fileFilter);
			if (entries != null) {
				for (File entry : entries)
					if (entry.isDirectory())
						crawl(entry);
					else if (!alreadyIndexed(entry)) {
						fileQueue.get(queueIndex).put(entry);
						queueIndex = (queueIndex + 1) % fileQueue.size();
					}
			}
		}
	}

	static class Indexer implements Runnable {
		private final List<BlockingDeque<File>> queue;
		private int myIndex;

		public Indexer(List<BlockingDeque<File>> queue, final int index) {
			this.queue = queue;
			this.myIndex = index;
		}

		public void run() {
			try {
				while (true) {
					File file = null;
					int currentIndex = myIndex;
					while (file == null) {
						file = (currentIndex == myIndex) ? queue.get(myIndex).poll(0, TimeUnit.NANOSECONDS)
														 : queue.get(currentIndex).pollLast(0, TimeUnit.NANOSECONDS);
						currentIndex = (currentIndex + 1) % queue.size();
					}
					indexFile(file);
				}
			} catch (InterruptedException e) {
				while (!queue.get(myIndex).isEmpty()) {
					indexFile(queue.get(myIndex).poll());
				}
			} finally {
				System.out.println("Indexer thread " + Thread.currentThread().getId() + " interrupted");
			}
		}

		public void indexFile(File file) {
			System.out.println(file.toString());
		};
	}

	private static final int BOUND = 10;
	private static final int N_CONSUMERS = Runtime.getRuntime().availableProcessors();

	public static void startIndexing(File[] roots) throws InterruptedException {
		List<BlockingDeque<File>> queueList = new ArrayList<BlockingDeque<File>>();
		CountDownLatch crawlerLatch = new CountDownLatch(roots.length);
		List<Thread> indexerThreadList = new ArrayList<Thread>();
		for (int i = 0; i < N_CONSUMERS; i++) {
			queueList.add(new LinkedBlockingDeque<File>(BOUND));
		}

		FileFilter filter = new FileFilter() {
			public boolean accept(File file) {
				return true;
			}
		};
		
		for (File root : roots) {
			new Thread(new FileCrawler(queueList, filter, root, crawlerLatch)).start();
		}

		int queueIndex = 0;
		for (int i = 0; i < N_CONSUMERS; i++) {
			Thread indexer = new Thread(new Indexer(queueList, queueIndex++));
			indexerThreadList.add(indexer);
			indexer.start();
		}

		try {
			crawlerLatch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		for (Thread indexer: indexerThreadList) {
			indexer.interrupt();
		}
	}

	public static void main(String[] args) throws InterruptedException {
		startIndexing(new File[] { new File("C:\\DB_0002"), new File("C:\\DB_0001")});
	}
}